package com.shujia.MapReduce;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.InputSplit;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.FileSplit;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import java.io.IOException;

public class Demo6ReduceJoin {
    public static class JoinMapper extends Mapper<LongWritable, Text, Text, Text> {
        @Override
        protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, Text>.Context context) throws IOException, InterruptedException {
            // 输入：1500100001,施笑槐,22,女,文科六班
            // 输出：1500100001 1500100001,施笑槐,22,女,文科六班#

            // 输入：1500100001,1000001,98
            // 输出：1500100001 1500100001,1000001,98*
            // 输入：1500100001,1000003,137
            // 输出：1500100001 1500100001,1000003,137*

            // 经过一系列操作最终输入Reduce：1500100001 {1500100001,施笑槐,22,女,文科六班#|1500100001,1000001,98*|1500100001,1000003,137*}

            // 获取数据的路径
            InputSplit inputSplit = context.getInputSplit();
            FileSplit fileSplit = (FileSplit) inputSplit;
            String path = fileSplit.getPath().toString();

            String[] splits = value.toString().split(",");
            // 取学生id 作为Map输出的key
            String id = splits[0];
            // Map输出的Value，根据文件来源的不同 给当行数据拼接不同的字符 # *
            String outV = value.toString();
            if (path.contains("student")) {
                // 给学生文件的数据打上 #
                outV += "#";
            } else if (path.contains("score")) {
                outV += "*";
            }
            context.write(new Text(id), new Text(outV));

        }


    }

    public static class JoinReducer extends Reducer<Text, Text, Text, Text> {
        // 1500100001 {1500100001,施笑槐,22,女,文科六班#|1500100001,1000001,98*|1500100001,1000003,137*}
        // 输出 id name,gender,sum_score
        @Override
        protected void reduce(Text key, Iterable<Text> values, Reducer<Text, Text, Text, Text>.Context context) throws IOException, InterruptedException {
            String name = null;
            String gender = null;
            int sum_score = 0;
            for (Text value : values) {
                String line = value.toString();
                String[] splits = line.split(",");
                if (line.endsWith("#")) {
                    // 处理学生数据
                    name = splits[1];
                    gender = splits[3];
                } else if (line.endsWith("*")) {
                    String score_str = splits[2].split("\\*")[0];
                    int score = Integer.parseInt(score_str);
                    sum_score += score;
                }
            }
            context.write(key, new Text(name + "," + gender + "," + sum_score));
        }
    }

    public static void main(String[] args) throws IOException, InterruptedException, ClassNotFoundException {
        Configuration conf = new Configuration();
        Job job = Job.getInstance(conf);

        job.setJobName("Demo6ReduceJoin");
        job.setJarByClass(Demo6ReduceJoin.class);

        // 分别对Map Reduce进行配置
        job.setMapperClass(JoinMapper.class);
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(Text.class);

        job.setReducerClass(JoinReducer.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(Text.class);

        // 设置输入输出路径
        // 需要提前将数据上传至HDFS对应的目录
        FileInputFormat.addInputPath(job, new Path("/data/student/"));
        FileInputFormat.addInputPath(job, new Path("/data/score/"));

        FileOutputFormat.setOutputPath(job, new Path("/joinOutput"));

        job.waitForCompletion(true);
    }

    /**
     *
     * hadoop jar Hadoop-1.0.jar com.shujia.MapReduce.Demo6ReduceJoin
     */
}
